package com.ad2campaign.engine.core.tasklet;

import java.util.Date;
import java.util.HashMap;
import java.util.HashSet;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.Timer;
import java.util.TimerTask;
import java.util.UUID;

import org.apache.log4j.Logger;
import org.springframework.batch.core.JobParameters;
import org.springframework.batch.core.JobParametersBuilder;

import com.ad2campaign.engine.core.EngineActions;
import com.ad2campaign.engine.core.TaskStatus;
import com.ad2campaign.engine.core.service.GenericJobService;
import com.ad2campaign.engine.db.dao.PushNotificationMasterDao;
import com.ad2campaign.engine.db.model.Node;
import com.ad2campaign.engine.db.model.PushNotificationMaster;
import com.ad2campaign.engine.rest.Loadbalancer;
import com.ad2campaign.engine.rest.RequestParameter;
import com.ad2campaign.engine.rest.RestActionResponse;

public class TableScanningTask extends TimerTask 
{
	private static final Logger LOGGER = Logger.getLogger(TableScanningTask.class);
 	private List<Thread> threads=null; 
	private PushNotificationMasterDao pushNotificationMasterDao;
	private Loadbalancer loadbalancer;
	private String classIOS;
 	private String jobname;
 	private String selfHost; 
 	private int  selfPort;
	private GenericJobService genericJobService;
 	private String classAndroid;
 	private List<RequestParameter> requestParameters = new LinkedList<RequestParameter>();
	private Map<String,String> urlVariables = new HashMap<String, String>();
	private long destroyWaitInterval=20000;
	private TimerTask recoveryTask;

 	
	
	public TimerTask getRecoveryTask() {
		return recoveryTask;
	}

	public void setRecoveryTask(TimerTask recoveryTask) {
		this.recoveryTask = recoveryTask;
	}

	@Override
	public void run() 
	{
		requestParameters.clear();
		urlVariables.put("jobName", jobname);
 		threads.add(Thread.currentThread());
		List<PushNotificationMaster> resultList = pushNotificationMasterDao.findPushByDate(new Date());
  		LOGGER.info("TableScanningTask.run():  Got ["+resultList.size()+"] new message(s) to process");
		if (resultList!=null)
		{	
			for(PushNotificationMaster pnm:resultList)
			{
				LOGGER.info("TableScanningTask.run():  Picked message of ID "+pnm.getId());
				if (LOGGER.isDebugEnabled())
				{
					LOGGER.debug("TableScanningTask.run(): Processing message ["+pnm.toString()+"]");
				}
				String parameter = parameterToString(pnm,classIOS,classAndroid);
				requestParameters.add(new RequestParameter("jobParam", parameter));
				RestActionResponse response= loadbalancer.process(EngineActions.START.getAction(), requestParameters, urlVariables,null);
				LOGGER.info("TableScanningTask.run(): Load balancer response "+response);
				
				Long executionId=Long.parseLong(String.valueOf(response.getData().get(0)));
 				if (response.isStatus())
				{
					LOGGER.info(String.format("TableScanningTask.run(): Job stared successfully with execution id [%d] ,Setting current status of task is [%s] ",executionId,TaskStatus.PICKED.getStatusString()));
					pushNotificationMasterDao.updateMessageStatusAndExecutionIdById(pnm.getId(), executionId, TaskStatus.PICKED.getStatusValue());
				}
				else
				{
					LOGGER.info(String.format("TableScanningTask.run(): Unable to start job setting failed execution id to [%d] ,Setting current status of task to [%s] ",-1,TaskStatus.READY.getStatusString()));
			 		pushNotificationMasterDao.updateMessageStatusAndExecutionIdById(pnm.getId(), -1L, TaskStatus.READY.getStatusValue());
				}
				
				LOGGER.info(String.format("TableScanningTask.run():  Completed processing of message of ID [%d] with response %s",pnm.getId(),response));
 			}
		}
	}
	
	private String parameterToString(PushNotificationMaster pnm,String IOSBean,String AndroidBean) 
	{

		JobParametersBuilder builder = new JobParametersBuilder();
		builder.addString("dummyid",UUID.randomUUID().toString());
		
		builder.addString("master_id",pnm.getId()+"");
		builder.addString("master_bId",pnm.getbId()+"");
		builder.addString("master_cId", pnm.getcId()+"");
		builder.addString("master_device", pnm.getDevice());
		builder.addString("master_where_clause",pnm.getWhere_clause());
		builder.addString("master_src_tablename",pnm.getSrc_tablename());
		builder.addString("master_dest_tablename",pnm.getDest_tablename());
		builder.addString("master_message", pnm.getMessage());
		builder.addString("master_key_val",pnm.getKey_val());
		
		if(pnm.getDevice().equalsIgnoreCase("ios")){
			builder.addString("pushNotificationService",IOSBean);//For IOS
		}else if(pnm.getDevice().equalsIgnoreCase("android")){
			builder.addString("pushNotificationService",AndroidBean);//For Android
		}
		
		JobParameters jobParameters = builder.toJobParameters();

 		 
 		//Removing [ ] 
		String parameter =jobParameters.toString();
		parameter=parameter.substring(1,parameter.length()-1);
		return parameter;
	}

	
	
	
	public int getSelfPort() {
		return selfPort;
	}

	public void setSelfPort(int selfPort) {
		this.selfPort = selfPort;
	}

	public String getJobname() {
		return jobname;
	}
	public String getSelfHost() {
		return selfHost;
	}

	public void setSelfHost(String selfHost) {
		this.selfHost = selfHost;
	}
	
	public void init()
	{
		Timer timer = new Timer();
		timer.schedule(recoveryTask, 5000);
	}
	
	public void destroy()
	{
		// Stop all jobs
		LOGGER.info("destroy():Stopping all jobs......");

		genericJobService.stopAllJob(jobname);
 		long destroyEndmili = System.currentTimeMillis()+ destroyWaitInterval;

 			while (destroyEndmili  >=System.currentTimeMillis()  )
 			{
 				RestActionResponse response =genericJobService.stopping(jobname); 
 				if (response.getData()!=null && response.getData().size() == 0 )
 				{
 					break;
 				}
 				LOGGER.info( String.format("destroy():Found [%d] jobs in stopping status, waiting for them to be stop ",response.getData().size()));
 				try 
 				{
					Thread.sleep(2000);
				} catch (InterruptedException e) {}
 			}
 		
		Node selfNode=null;
		List<Node> nodes = loadbalancer.getNodesLoader().loadNodes();
		for (Node node :nodes)
		{
			if (node.getHost().equalsIgnoreCase(selfHost) && node.getPort()==selfPort)
			{
				selfNode =  node;
				break;
			}
		}

		Set<Node> exclude = new HashSet<Node>();
		exclude.add(selfNode);
 		LOGGER.info("destroy():Sending jobs to other nodes...");
		genericJobService.resumeJobs(exclude,jobname);
		LOGGER.info("destroy():Stopping loadbalancer");
		loadbalancer.destroy();
	
	}



	public GenericJobService getGenericJobService() {
		return genericJobService;
	}

	public void setGenericJobService(GenericJobService genericJobService) {
		this.genericJobService = genericJobService;
	}

	public void setJobname(String jobname) {
		this.jobname = jobname;
	}


	public long getDestroyWaitInterval() {
		return destroyWaitInterval;
	}

	public void setDestroyWaitInterval(long destroyWaitInterval) {
		this.destroyWaitInterval = destroyWaitInterval;
	}

	public String getClassIOS() {
		return classIOS;
	}

	public void setClassIOS(String classIOS) {
		this.classIOS = classIOS;
	}

	public String getClassAndroid() {
		return classAndroid;
	}

	public void setClassAndroid(String classAndroid) {
		this.classAndroid = classAndroid;
	}

	public List<Thread> getThreads() {
		return threads;
	}

	public void setThreads(List<Thread> threads) {
		this.threads = threads;
	}

	public PushNotificationMasterDao getPushNotificationMasterDao() {
		return pushNotificationMasterDao;
	}

	public void setPushNotificationMasterDao(
			PushNotificationMasterDao pushNotificationMasterDao) {
		this.pushNotificationMasterDao = pushNotificationMasterDao;
	}

	public Loadbalancer getLoadbalancer() {
		return loadbalancer;
	}

	public void setLoadbalancer(Loadbalancer loadbalancer) {
		this.loadbalancer = loadbalancer;
	}

	public TableScanningTask(List<Thread> threads)
	{
		this.threads=threads;
	}

}
